package com.wang.wx.service.impl;

import com.google.common.hash.BloomFilter;
import com.wang.wx.config.BloomFilterConfig;
import com.wang.wx.constants.BatchConstants;
import com.wang.wx.constants.ThreadPoolManager;
import com.wang.wx.entity.TwitterUser;
import com.wang.wx.entity.WxUser;
import com.wang.wx.service.DataSplitterService;
import com.wang.wx.service.TwitterUserService;
import com.wang.wx.service.WxUserService;
import com.wang.wx.utils.DynamicDataSplitterUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.util.concurrent.BlockingQueue;

/**
 * @author wjh
 * @create 2024-08-02 15:59
 */
@Service
@Slf4j
public class TwitterDataSplitterServiceImpl implements DataSplitterService {

    @Value("${file.batch.size}")
    private int batchSize;
    @Resource
    private BloomFilterConfig bloomFilterConfig;
    @Value("${file.thread.pool.size}")
    private int threadPoolSize;
    @Resource
    private TwitterUserService twitterUserService;

    @Override
    public void handleFile(String file) throws IOException {
        initPool();
//        BloomFilter<String> bloomFilter = bloomFilterConfig.createBloomFilter();
        BlockingQueue<TwitterUser> queue = BatchConstants.TWITTER_QUEUE;
        BatchConstants.TWITTER_SIGNAL_COMPLETION.set(false);

        try (BufferedReader reader = new BufferedReader(new FileReader(file))) {
            String line;
            while ((line = reader.readLine()) != null) {
                TwitterUser twitterUser = DynamicDataSplitterUtil.twitterUserBloomFilter(line.trim());
                if (twitterUser == null) {
                    continue;
                }
                queue.put(twitterUser);
            }
        } catch (IOException | InterruptedException e) {
            log.error("读取文件时发生错误：{}", e.getMessage());
            Thread.currentThread().interrupt();
            throw new IOException("读取文件时发生错误", e);
        } finally {
            BatchConstants.TWITTER_SIGNAL_COMPLETION.set(true);
            log.info("入库完成");
            ThreadPoolManager.shutdown();
            log.info("线程池关闭");
        }
    }

    public void initPool() {
        ThreadPoolManager.getExecutor(threadPoolSize);
        for (int i = 0; i < threadPoolSize; i++) {
            ThreadPoolManager.submitTask(() -> BatchProcessor.processTwitterQueue(twitterUserService, batchSize));
        }
    }
}
